package com.dw.sprboosoc.service;

import com.alibaba.fastjson.JSON;
import com.dw.sprboosoc.constant.MessageEnum;
import com.dw.sprboosoc.dto.WebSocketMessageDto;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * WebSocket 核心类
 * dingwen
 * 2021/2/20 10:41
 **/
// 添加Bean
@Component
@Slf4j
//访问路径
@ServerEndpoint(value = "/websocket/{sendUserId}")
public class WebSocketServer {
    //当前在线连接数，保证线程安全
    private static final AtomicInteger currentOnlineNumber = new AtomicInteger();
    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象
    private static final ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    //设置为静态的 公用一个消息map ConcurrentMap为线程安全的map  HashMap不安全
    private static final ConcurrentMap<String, Map<String, List<WebSocketMessageDto>>> messageMap = new ConcurrentHashMap<>();

    /*
     *发送消息
     * @param [session, message, userId]
     * @return void
     */
    public void sendMessage(WebSocketMessageDto webSocketMessageDto) throws IOException {
        try {
            switch (webSocketMessageDto.getMessageEnum()) {
                // 广播消息
                case ALL:
                    sessionPool.values().forEach(se -> {
                        try {
                            se
                                    .getBasicRemote()
                                    .sendText(webSocketMessageDto.toString());
                        } catch (IOException e) {
                           log.info("群发");
                        }
                    });
                    log.info("websocket: 广播消息" + webSocketMessageDto);
                    // 离线
                    storeOfflineMessage(webSocketMessageDto);
                    break;
                // 私发
                case ONE:
                    if (judgeUserOnline(webSocketMessageDto.getRecvUserId())) {
                        // 在线
                        sessionPool.get(webSocketMessageDto.getRecvUserId())
                                .getBasicRemote()
                                .sendText(webSocketMessageDto.toString());
                    } else {
                        // 离线
                        storeOfflineMessage(webSocketMessageDto);
                    }

                    log.info("websocket: 私发消息" + webSocketMessageDto);
                    break;
            }


        } catch (Exception exception) {
            log.error("websocket: 发送消息发生了错误");
        }
    }

    /*
     *客户端收到消息
     * @param [message]
     * @return void
     */
    @OnMessage
    public void onMessage(String webSocketMessageDtoStr) throws IOException {
        WebSocketMessageDto webSocketMessageDto = JSON.parseObject(webSocketMessageDtoStr, WebSocketMessageDto.class);
        log.info("websocket:" + webSocketMessageDto.getRecvUserId() + "收到，来自" + webSocketMessageDto.getSendUserId() + "发送的消息" + webSocketMessageDto.getMessage());
        sendMessage(webSocketMessageDto);
    }

    /*
     *判断用户是否在线
     * @param [recvUserId]
     * @return boolean
     */
    public boolean judgeUserOnline(String recvUserId) {
        boolean flag = !ObjectUtils.isEmpty(sessionPool.get(recvUserId));
        String flagStr = flag ? "在线" : "离线";
        log.info("websocket: " + recvUserId + ":" + flagStr);
        return flag;
    }

    /*
     *用户离线时把消息存储到内存
     * @param [recvUserId]
     * @return void
     */
    public void storeOfflineMessage(WebSocketMessageDto webSocketMessageDto) {
        //用户不在线时 第一次给他发消息
        if (ObjectUtils.isEmpty(messageMap.get(webSocketMessageDto.getRecvUserId()))) {
            Map<String, List<WebSocketMessageDto>> maps = new HashMap<>();
            List<WebSocketMessageDto> list = new ArrayList<>();
            list.add(webSocketMessageDto);
            maps.put(webSocketMessageDto.getRecvUserId(), list);
            messageMap.put(webSocketMessageDto.getRecvUserId(), maps);
        } else {
            //用户不在线时 再次发送消息
            Map<String, List<WebSocketMessageDto>> listObject = messageMap.get(webSocketMessageDto.getRecvUserId());
            List<WebSocketMessageDto> objects = new ArrayList<>();
            if (!ObjectUtils.isEmpty(listObject.get(webSocketMessageDto.getRecvUserId()))) {//这个用户给收消息的这个用户发过消息
                //此用户给该用户发送过离线消息（此用户给该用户发过的所有消息）
                objects = listObject.get(webSocketMessageDto.getRecvUserId());
                //加上这次发送的消息
                objects.add(webSocketMessageDto);
                //替换原来的map
                listObject.put(webSocketMessageDto.getRecvUserId(), objects);
            } else {//这个用户没给该用户发送过离线消息
                objects.add(webSocketMessageDto);
                listObject.put(webSocketMessageDto.getRecvUserId(), objects);
            }
            messageMap.put(webSocketMessageDto.getRecvUserId(), listObject);


        }
    }

    /*
     *成功建立连接后调用
     * @param [session, userId]
     * @return void
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "sendUserId") String sendUserId) throws IOException {
        //成功建立连接后加入
        sessionPool.put(sendUserId, session);
        //当前在线数量+1
        currentOnlineNumber.incrementAndGet();
        log.info("websocket:" + sendUserId + "加入连接,当前在线用户" + currentOnlineNumber + "未读消息数：" + getMessageCount(sendUserId));
        // 发送离线消息
        sendOffLineMessage(sendUserId);
    }

    /*
     * 用户上线时发送离线消息
     * @param []
     * @return void
     */
    @SneakyThrows
    public void sendOffLineMessage(String sendUserId) {

        if (ObjectUtils.isEmpty(messageMap.get(sendUserId))) {
            // 该用户没有离线消息
            return;
        }
        // 当前登录用户有离线消息
        //说明在用户没有登录的时候有人给用户发送消息
        //该用户所有未收的消息
        Map<String, List<WebSocketMessageDto>> lists = messageMap.get(sendUserId);
        //对象用户发送的离线消息
        List<WebSocketMessageDto> list = lists.get(sendUserId);
        if (list != null) {
            for (WebSocketMessageDto webSocketMessageDto : list) {
                onMessage(JSON.toJSONString(webSocketMessageDto));
            }
        }

        // 删除已发送的消息
        removeHasBeenSentMessage(sendUserId, lists);


    }

    /*
     *删除已发送的消息
     * @param [sendUserId, map]
     * @return void
     */
    public void removeHasBeenSentMessage(String sendUserId, Map<String, List<WebSocketMessageDto>> map) {
        // map中key（键）的迭代器对象
        //用户接收完消息后删除 避免下次继续发送
        Iterator iterator = map.keySet().iterator();
        while (iterator.hasNext()) {// 循环取键值进行判断
            String keys = (String) iterator.next();//键
            if (sendUserId.equals(keys)) {
                iterator.remove();
            }
        }
    }

    /*
     *关闭连接时调用
     * @param [userId]
     * @return void
     */
    @OnClose
    public void onClose(@PathParam(value = "sendUserId") String sendUserId) {
        sessionPool.remove(sendUserId);
        currentOnlineNumber.decrementAndGet();
        log.info("websocket:" + sendUserId + "断开连接,当前在线用户" + currentOnlineNumber);
    }

    /*
     *发生错误时调用
     * @param [session, throwable]
     * @return void
     */
    @OnError
    public void onError(Throwable throwable) {
        log.error("websocket: 发生了错误");
        throwable.printStackTrace();
    }

    /**
     * 获取该用户未读的消息数量
     */
    public int getMessageCount(String recvUserId) {
        //获取该用户所有未收的消息
        Map<String, List<WebSocketMessageDto>> listMap = messageMap.get(recvUserId);
        if (listMap != null) {
            List<WebSocketMessageDto> list = listMap.get(recvUserId);
            if (list != null) {
                return listMap.get(recvUserId).size();
            } else {
                return 0;
            }

        } else {
            return 0;
        }

    }
}
